package com.ctyun.xstore.vod;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ctyun.xstore.vod.video.model.PartETag;

public class ConcurrentPartUploader extends AbstractPartUploader {

	private final Logger log = LoggerFactory.getLogger(ConcurrentPartUploader.class);

	private static final int DEFAULT_MAX_POOL_SIZE = 4;

	private final ExecutorService executeService;

	public ConcurrentPartUploader() {
		this(DEFAULT_MAX_POOL_SIZE);
	}

	public ConcurrentPartUploader(int maxPoolSize) {
		executeService = new ThreadPoolExecutor(0, maxPoolSize, 0, TimeUnit.MILLISECONDS,
				new LinkedBlockingQueue<Runnable>());
	}

	@Override
	protected List<PartETag> doMultiPartupload(File file, List<String> urls, int partSize) throws IOException {
		log.info("Start to multipartUpload concurrently....");
		try (FileInputStream fin = new FileInputStream(file); FileChannel channel = fin.getChannel()) {
			ByteBuffer buffer = ByteBuffer.allocate(partSize);
			int index = 0;
			int len = 0;
			List<Future<String>> resultList = new ArrayList<>();
			while ((len = (channel.read(buffer))) != -1) {
				final byte[] array = new byte[len];
				((Buffer) buffer).flip();
				buffer.get(array);
				final int indexFinal = index;
				Future<String> result = this.executeService.submit(() -> this.upload(urls.get(indexFinal), array));
				resultList.add(result);
				((Buffer) buffer).clear();
				index++;
			}

			List<PartETag> etags = new ArrayList<>(resultList.size());
			for (int i = 0; i < resultList.size(); i++) {
				try {
					Future<String> future = resultList.get(i);
					PartETag etag = new PartETag(i + 1, future.get());
					etags.add(etag);
				} catch (Exception e) {
					throw new IllegalStateException("ThreadPoolExecutor error:" + e.getMessage());
				}
			}
			log.info("MultipartUpload concurrently has done!");
			return etags;
		}
	}
}
